package com.iflytek.hive.step1mapjoin;


import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

public class MapJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    private HashMap<String, String> tbSaleMap = new HashMap<>();
    private HashMap<String, String> tbDateMap = new HashMap<>();
    private final Text keyOut = new Text();
    private final Text valueOut = new Text();
    // 第一件事情:将分布式缓存的小表数据读取到本地Map集合(只需要做一次)
    @Override
    protected void setup(Context context) throws IOException, InterruptedException
    {
        // 1:获取分布式缓存文件列表
        URI[] cacheFiles = context.getCacheFiles();
        // 2:通过缓存文件获取其路径
        String tbSalePath = cacheFiles[0].getPath().toString();
        String tbDatePath = cacheFiles[1].getPath().toString();
        // 4:读取文件内容, 并将数据存入Map集合
        // 4.1 将字节输入流转为字符缓冲流FSDataInputStream --->BufferedReader
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(tbSalePath),"UTF-8"));
        // 4.2 读取小表文件内容,以行位单位,并将读取的数据存入map集合
        String line = null;
        while ((line = bufferedReader.readLine()) != null) {
            String[] split = line.split(",");
            tbSaleMap.put(split[0], line);
        }

        // 4:读取文件内容, 并将数据存入Map集合
        // 4.1 将字节输入流转为字符缓冲流FSDataInputStream --->BufferedReader
        BufferedReader bufferedReader1 = new BufferedReader(new InputStreamReader(new FileInputStream(tbDatePath),"UTF-8"));
        // 4.2 读取小表文件内容,以行位单位,并将读取的数据存入map集合
        String line1 = null;
        while ((line1 = bufferedReader1.readLine()) != null) {
            String[] split1 = line1.split(",");
            tbDateMap.put(split1[0], line1);
        }
        // 5:关闭流
        bufferedReader.close();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 将行文本数据切分
        String[] fields = value.toString().split(",");
        // 获取订单ID
        String orderId = fields[0];
        // 根据订单ID在tbSaleMap缓存集合中找出来对应的订单信息
        String orderInfo = tbSaleMap.get(orderId);
        // 将订单信息数据切分
        String[] orderInfoArr = orderInfo.split(",");
        // 获取订单时间
        String orderDate = orderInfoArr[2];
        // 根据订单时间在tbDateMap时间维度表中找出对应的时间信息
        String dateInfo = tbDateMap.get(orderDate);
        // 将时间信息数据切分
        String[] dateInfoArr = dateInfo.split(",");
        // 获取时间信息的季度
        String dateQuat = dateInfoArr[7];
        // 拼接数据
        String valueLine = orderDate + "," + dateQuat +  "," + value.toString();
        keyOut.set(valueLine);
        context.write(keyOut, NullWritable.get());

    }

}
